/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.dtub.mongo.update.worker;

import cn.ac.iie.di.ban.dtub.mongo.update.util.MongoHelper;
import cn.ac.iie.di.ban.dtub.mongo.update.util.SenderPool;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import static java.lang.Thread.sleep;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class Sender extends Thread {

    static private final Logger logger = Logger.getLogger(Sender.class.getName());

    REReceiveSession session;
    ThreadPoolExecutor threadPool;
    ArrayBlockingQueue<HashMap<String, Object>> queue;
    private final MongoHelper[] mongos;
    private final SenderPool[] pools;

    public Sender(SenderPool[] senderPools, MongoHelper[] mongos, ArrayBlockingQueue<HashMap<String, Object>> queue) {
//        this.handler = new SenderHandler(senderPools, ssPool, mongos, queue);
        this.mongos = mongos;
        this.pools = senderPools;
        this.threadPool = new ThreadPoolExecutor(16, 20, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(4), new ThreadFactory() {

            String threadGroupName = "LoadProcessSenderThread - ";
            AtomicInteger count = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, threadGroupName + count.getAndIncrement());
            }
        }, new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                while (true) {
                    try {
                        executor.getQueue().put(r);
                        break;
                    } catch (InterruptedException ex) {
                    }
                }
            }
        });
        this.queue = queue;
    }
    AtomicInteger count = new AtomicInteger(0);

    @Override
    public void run() {
        logger.info("sender is started!");
        try {
            sleep(500);
        } catch (InterruptedException ex) {
        }
        while (true) {
            if (queue.isEmpty()) {
                try {
                    logger.info("list empty! the remaining capacity of queue is: " + queue.remainingCapacity());
                    sleep(500);
                } catch (InterruptedException ex) {
                } catch (Exception e) {

                }
            } else {
                HashMap<String, Object> urlMap = new HashMap<>(200);
                for (int i = 0; i < 200; ++i) {
                    try {
                        HashMap<String, Object> line = queue.poll(3, TimeUnit.SECONDS);
                        if (line == null) {
                            break;
                        }
                        if (urlMap.containsKey((String) line.get("_id"))) {
//                            logger.info(line.get("_id") + "---" + line.get("nrd"));
                            Long nrd = (Long) line.get("nrd");
                            Long nrd1 = (Long) ((HashMap) urlMap.get((String) line.get("_id"))).get("nrd");
                            Long nrdSum = nrd + nrd1;
                            line.put("nrd", nrdSum);
                            urlMap.put((String) line.get("_id"), line);
                        } else {
                            urlMap.put((String) line.get("_id"), line);
                        }

                    } catch (InterruptedException ex) {
                        break;
                    }
                }
                if (count.incrementAndGet() > 50) {
                    count.set(0);
                    logger.info("the remaining capacity of queu is: " + queue.remainingCapacity());
                }
                if (urlMap.isEmpty()) {
                    try {
                        logger.info("list empty! the remaining capacity of queue is: " + queue.remainingCapacity());
                        sleep(50);
                    } catch (InterruptedException ex) {
                    } finally {
                        continue;
                    }
                } else {
                    logger.info("list size: " + urlMap.size() + " the remaining capacity of queue is: " + queue.remainingCapacity());
                    threadPool.execute(new SenderHandler(pools, mongos, urlMap));

                }
            }
        }
    }

}
